package org.example

import org.apache.spark.sql.SparkSession

object ketang1 {
  def main(args: Array[String]): Unit = {
    //创造运行spark环境
    val spark =SparkSession.builder().master("local[*]").appName("spark").getOrCreate()
    val sc = spark.sparkContext
    //1.创建rdd 不指定分区就会打散
    val rdd1 =sc.makeRDD(List(1,5,3,7,6))
    val rdd2 =sc.parallelize(List(('a',90),('b',78),('c',66),('b',78)))
    //2.spark数据分析方法
    val rdd3 =sc.makeRDD(List(1,2,3,4,5),1)

    /*//交集 intersection
    rdd1.intersection(rdd3).foreach(println)
    //并集 union
    rdd1.union(rdd3).foreach(println)
    //差集 subtract 站在rdd1的角度上3有1没有的
    rdd1.subtract(rdd3).foreach(println)
    //zip拉链 分区数要一致 形成元组（key value）
    rdd1.zip(rdd3).foreach(println)
    //笛卡尔积 组合新的元组5*5
    rdd1.cartesian(rdd3).foreach(println)
    //filter rdd2中筛选出大于等于90的数据
    rdd2.filter(tp => tp._2 >=90).foreach(println)
    rdd2.filter(_._2 >=90).foreach(println)//tp只出现一次可省略
    //去重
    rdd2.distinct().foreach(println)
    //3.输出结果
    rdd1.foreach(println)
    rdd2.collect().foreach(println)*/

    //键值树（key value）
    //reduceByKey   聚合
    val data1 =sc.makeRDD(List(("a",90),('b',78),('c',66),('b',78)),2)
    val data2 =sc.makeRDD(List(("a",5),('b',3),('c',4),('a',6)),2)
    data1.reduceByKey((x,y)=> x + y).foreach(println)
    //groupByKey  根据Key分组
    //data1.groupByKey().foreach(println)
    //data1.groupBy(_._1).foreach(println)
    //join  相同的key和value组成新的元组 如果没有匹配上就不会出现 如果存在多个相同的key 会依次匹配
    //data1.join(data2).foreach(println)
    //以左边为主，右边可能匹配不到
    //data1.leftOuterJoin(data2).foreach(println)
    //combinByKey  联合 适用于需要对相同键的值进行聚合
    data1.combineByKey(
      v =>(v,1),
      (t:(Int,Int),v) =>{
        (t._1+v,t._2+1)
      },
      (t1: (Int, Int), t2:(Int,Int)) => {
        (t1._1+t2._1,t1._2+t2._2)
      }
    ).foreach(println)//重要*****
    //集合 适用于需要在每个分区进行聚合 计算每个键的最大值和总和
    data1.aggregateByKey(0)(math.max(_,_),_+_).foreach(println)
    //结束后释放资源
    sc.stop()
  }

}
